11. 采集路径并发安全与恢复能力设计

版本：v1.0　范围：internal/core（ScanEngine / ChannelManager / ExecutionLayer / ShadowCore）+ internal/driver（13 协议驱动 + ConnectionManager）目标：消除采集链路上的死锁与数据竞态，并建立工业级故障恢复闭环（异常检测 → 隔离 → 探测 → 恢复）。原则：只做收敛与加固，不以牺牲稳定性换取吞吐。

0. 结论摘要（TL;DR）

类别	修复前	修复后
确定性死锁	EtherNet/IP 重连必然自锁死（`t.mu` 重入）	消除（`resetConnection` 独立临界区）
并发重复执行	防饿死对已在队列的任务重复入堆 → 同一任务双 worker 并发执行	消除（`ScanTask.queued` 队列成员位 + `heap.Fix`）
进程级崩溃	驱动 panic 在 worker goroutine 中无 recover → 整进程退出	消除（`readPoints` + `executeTaskAsync` + 调度/巡检循环 recover）
并发 map 崩溃	`task.Params` 原地改写 vs 热路径无锁读 → `fatal error: concurrent map read and map write`	消除（Params copy-on-write）
通道卡死无恢复	链路断开后无周期性重连 → 只能重启进程	消除（通道健康看门狗 + 冷却）
大锁阻塞	OPC UA / BACnet 持驱动全局锁做网络 RPC	收敛（快照后释放锁再 I/O）

1. 采集路径清单（逐一列出）

采集数据从调度到影子/管道共 6 层。下表是全部路径及其写者/读者。

1.1 端到端主链路

#	阶段	位置	并发执行者	关键共享状态
P1	任务注册	`ScanEngine.addTask`	ChannelManager（持 `cm.mu`）	`tasks` map、`priorityQueue`
P2	到期判定/出队	`dispatchLoop → processReadyTasks → popReadyTaskEDF`	ScanEngine dispatch 单协程	`priorityQueue`、`task.Status`
P3	抖动钳制/防饿死	`enforceHardJitterClamp`、`enforceAntiStarvation`	dispatch 单协程	`priorityQueue`、`overdueWarnAt`
P4	执行（异步）	`go executeTaskAsync`	每任务 1 协程（受 `ResourceController` 限流）	`ResourceController` 计数
P5	协议 I/O	`ExecutionLayer.Execute → execute{Serial,Parallel,Limited} → readPoints → driver.ReadPoints`	SerialQueue 每设备 1 协程 / WorkerPool N 协程	`channelMu`（共享链路）、驱动内部锁
P6	结果落地	`applyCollectToShadow → ShadowIngress → ShadowCore(COW) → ShadowBridge → DataPipeline`	入口协程 + flush 协程	`bufferMu`、`sc.mu`、`dp.mu`
P7	设备/通道状态回写	`finalizeScanCollect → NodeStatus / markChannelDevicesOffline`	入口协程 / API 协程	`node.Runtime`、`stateManager`

1.2 协议驱动路径（13 类）

协议	驱动结构体	传输层锁	共享链路串行化	重连入口	恢复机制
`bacnet-ip`	`BACnetDriver`	`mu RWMutex` + `DeviceContext.ReadMu/CacheMu`	channelMu（已绑定）	`ScheduleReconnect` + `scheduleDeviceRecovery`	通道级 + 设备级隔离/探测
`modbus-tcp/rtu/rtu-over-tcp`	`ModbusDriver`	`ModbusTransport.mu Mutex` + `PointScheduler.mu`	channelMu（已绑定）	`ScheduleReconnect`	点位冷却 SKIPPED + 链路错误豁免
`dlt645`	`DLT645Driver`	`DLT645Transport.mu Mutex`	channelMu（已绑定）	`scheduleReconnect`	冷却 + 发送节流
`knxnet-ip`	`KNXnetIPDriver`	`KNXTransport.mu/writeMu/cacheMu`	channelMu（已绑定）	`connectOnce`	心跳循环
`opc-ua`	`OpcUaDriver`	`mu RWMutex` + `ClientWrapper.mu` + `DeviceSubscription.mu`	驱动内锁	`onReconnectNeeded` → `scheduleReconnect`	订阅重连
`s7`	`S7Driver`	`S7Transport.mu Mutex`	—	`scheduleReconnect`	主动探针（唯一生产启用）
`ethernet-ip`	`EtherNetIPDriver`	`ENIPTransport.mu Mutex`	—	`scheduleReconnect`	失败计数触发
`ethercat`	`EtherCATDriver`	`cycleRunning` + `sync.Map` + `rxBuffer.mu`	—	`connectOnce`	PDO 周期线程 `cycleWG`
`ice104`	`ICE104Driver`	`writeMu` + `cacheMu` + `readWG`	—	`connectOnce`	读循环 cancel + WaitGroup
`profinet-io`	`ProfinetIODriver`	`ProfinetTransport.mu` + `ProfinetRPC.mu`	—	手动退避循环	`ensureDeviceConnection`
`mitsubishi-slmp`	`MitsubishiDriver`	`MCTransport.mu Mutex`	未绑定	—	无连接管理器
`omron-fins`	`OmronFinsDriver`	`tcpBackend.mu` / `udpBackend.mu`	未绑定	—	无连接管理器
`snmp`	`SNMPDriver`	`atomic` 计数	未绑定	—	无连接管理器

已知缺口（未在本次修复）：omron-fins / mitsubishi-slmp / snmp 在 isSharedLinkProtocol 中被判定为共享链路，但既未实现 drv.LinkMutexBinder，其 transport 也没有 ConnectionManager → 其同步 Connect 拨号不受 channelMu 串行化。见 §6 改进建议。

2. 锁与并发控制机制

2.1 锁清单（类型 / 粒度 / 保护对象）

层	锁	类型	粒度	保护对象
L1	`ChannelManager.mu`	`RWMutex`	全局拓扑	`channels`、`drivers`、`driverMus`、`jobs`
L1’	`ChannelManager.topologyDebounceMu`	`Mutex`	拓扑去抖	`topologyDebounceTimer`
L2	`ScanEngineAdapter.mu`	`Mutex`	设备注册	`driverManager`、`started`
L3	`ScanEngine.mu`	`RWMutex`	调度器	`tasks`、`priorityQueue`、`running`、`overdueWarnAt`、`taskIDCounter`
L4	`ScanTask.mu`	`RWMutex`	单任务	`Status`、`NextRun`、`DeadlineAt`、`Priority`、`Interval`、`queued`、`Params`(COW)
L5	`ExecutionLayer.mu`	`RWMutex`	执行层注册表	`protocolRegistry`、`driverRegistry`、`pointDegradation`、`ioProfileProvider`
L6	`SerialQueueManager.mu`	`RWMutex`	共享链路队列	`contexts`
L7	`ResourceController`	`int32` + atomic	并发上限	`goroutineCount`、`connectionCount`
L8	`ConnectionManager.mu / bgMu`	`Mutex`	连接状态机	`state`、`retryCount`、`cooldown`、`linkMu`、`bgCancel/bgDone`
L9	`driverMus[channelID]`（=channelMu）	`Mutex`	单通道	共享链路拨号 + I/O 串行化
L10	驱动内部锁	`Mutex/RWMutex`	单驱动	连接指针、缓存、统计
L11	`ShadowCore.mu` + `atomic.Pointer`	`Mutex` + COW	影子设备	快照发布（不可变）
L12	`ShadowIngress.bufferMu`	`Mutex`	影子入口缓冲	ring buffer 批次
L13	`DataPipeline.mu`	`Mutex`	管道	批次交换
L14	`NodeStatus.node.mu`	`Mutex`	单设备状态	`Runtime`

2.2 加锁顺序（强制，不可逆）

cm.mu(W)  →  adapter.mu  →  se.mu  →  task.mu
                                      ↑
                            (所有 heap 操作都在 se.mu 内，
                             Less/Swap 内部取 task.mu.RLock)

旁路（不得与主链交叉持有）：
  driverMus(channelMu)  →  driver.mu        ← runConnect 与 scan 读都遵守
  feedbackPendingMu     →  task.mu
  ShadowCore.mu         →  optimizer.mu

三条铁律（本次修复后的代码约束）：

sync.RWMutex 不可重入：已持 task.mu.Lock() 的路径禁止调用会再次取 task.mu 的方法。为此新增 taskShadowChannelIDLocked / taskDegradeOnFailureLocked / 直读 task.Params 的「Locked 变体」。
持 channelMu 期间只做拨号与 I/O，不得回调 ChannelManager 持 cm.mu 的方法（BACnet notifier 已用 applyDiscoveredDeviceLocked 规避）。
锁内不做网络 I/O：OPC UA / BACnet 已改为「加锁取快照 → 解锁 → 网络调用 → 回锁提交」。

2.3 并发原语选择理由

场景	采用	理由
任务状态字段	`RWMutex`	读多写少（每 tick 读、每轮写）
队列成员位 `queued`	随 `se.mu` 保护 + `task.mu` 短持	所有 heap 变更已在 `se.mu` 下，标志位与堆操作同临界区才可靠
连接状态/计数	`atomic.Bool/Int32/Int64`	无锁热路径
影子快照	COW + `atomic.Pointer`	读者永不见中间态，无锁读取
并发上限	atomic 计数	避免锁竞争
Params 配置	copy-on-write	消除「并发 map 读写」致命错误

3. 竞态条件场景（触发条件 + 复现路径）

C1　防饿死重复入队 → 同一任务并发双执行　【已修复】

触发条件：任务被 CanExecute()==false 回推入堆后，其 NextRun 逾期 > AntiStarvationSec（默认 300s）。
复现路径：enforceAntiStarvation 遍历 se.tasks（含已在堆中的任务）→ 无条件 heap.Push → 同一 *ScanTask 出现两份堆条目 → popReadyTaskEDF 对 Running 状态不做过滤 → 两个 worker 并发执行同一任务 → 竞写 task.Status / task.Points[i].DeviceID / rescheduleTask。
复现测试：TestAntiStarvation_RepeatedRescueKeepsSingleQueueEntry（断言队列长度恒为 1）。
修复：新增 ScanTask.queued；救援时若已在队列 → heap.FixLocked，否则 Push 并置位。

C2　EtherNet/IP 重连自死锁　【已修复】

触发条件：collectFailCount >= maxFailCount 触发 scheduleReconnect，且拨号成功。
复现路径：重连闭包 t.mu.Lock() → defer Unlock → 调 t.connectOnce → 内部再次 t.mu.Lock()（安装连接）→ 不可重入 Mutex 永久阻塞，t.mu 永不释放 → 后续 Read/Write/Disconnect 全部冻结，只能重启进程。
复现测试：TestENIPTransport_ScheduleReconnectDoesNotHoldLockWhileDialing（在拨号 factory 内 TryLock 探测，持有则失败）。
修复：抽出 resetConnection()（自持锁、快返回），在闭包中先复位再调 connectOnce。

C3　`task.Params` 并发 map 读写 → 进程崩溃　【已修复】

触发条件：UpdateTaskDriverConfig（BACnet 地址自动发现 / API）与扫描执行并发。
复现路径：写者持 se.mu 原地改写 task.Params["driverConfig"]；读者（execution_layer.readPoints / taskShadowChannelID / taskDegradeOnFailure）无锁读同一 map → Go 运行时 fatal error: concurrent map read and map write（不可 recover）。
复现测试：TestUpdateTaskDriverConfig_ConcurrentWithReaders（-race）。
修复：Params 全量 copy-on-write，setParams 原子发布；读者经 paramsSnapshot() 取不可变引用。

C4　驱动 panic 击穿进程　【已修复】

触发条件：驱动 ReadPoints 内 panic（畸形帧、越界、CGO 段错误）。
复现路径：executeSerial/Parallel 把 ReadFunc 提交到 SerialQueueManager / WorkerPool 的 worker 协程 → 该协程无 recover → panic 逃逸 → 进程退出，丢失全部内存态（熔断/降级/隔离）。
复现测试：TestExecutionLayer_RecoversDriverPanic、TestExecuteTaskAsync_RecoversPanicAndRearmsTask。
修复：在真正的执行边界 ExecutionLayer.readPoints 加 recover（覆盖 Serial/Parallel/Limited 全部路径），把 panic 转成普通读错误交给熔断器；executeTaskAsync 再兜底 recover 并重新武装任务。

C5　调度循环 panic → 全局停摆　【已修复】

触发条件：processReadyTasks 内部 panic（堆结构被破坏、指标异常）。
复现路径：dispatchLoop 是唯一调度协程，无 recover → 协程死亡 → 所有设备永久停止采集（直到进程重启）。
修复：safeProcessReadyTasks recover + rebuildQueueAfterPanic（从 se.tasks 重建堆，丢弃重复/悬垂指针，重算 queued）——即状态回滚。

C6　通道链路断开后无周期性重连　【已修复】

触发条件：驱动通信层半挂（如 BACnet UDP socket 未重启），Health()==Bad，但驱动自身未触发重连。
复现路径：isChannelLinkUp==false 仅被被动查询；core 层只在 StartChannel 失败与 finalizeScanCollect 链路错时动作 → 通道长期 offline，只能人工重启进程。
复现测试：TestChannelHealthWatchdog_TriggersReconnectWithCooldown。
修复：ChannelManager 新增 15s 健康看门狗；链路 down 且距上次尝试 ≥30s 时，对实现 ReconnectScheduler 的驱动触发 ScheduleReconnect。

C7　OPC UA / BACnet 持全局锁做网络 RPC　【已收敛】

触发条件：重连或懒连期间。
后果：d.mu 被持 10s（OPC UA connect timeout），期间 Health()/GetMetrics()/其他设备读全部阻塞。
修复：改为「加锁取快照 → 解锁 → Connect(ctx) → 回锁提交」。

C8　其余已识别但本次未改动的风险（登记在案）

编号	位置	风险
R1	`bacnet.Health()`	无锁读 `d.connected/d.client`（与写者竞态）
R2	`bacnet.connectOnce`	持 `d.mu` 调 `StopBackgroundLoop`，锁序与 `Disconnect` 不一致
R3	`NodeStatus.MarkOffline/onCollectUnreachable`	未持 `node.mu`，与 `onCollectSuccess` 保护不一致
R4	`DataPipeline.Stop`	与 `PushBatch` 交错存在 `send on closed channel` 窗口
R5	`ScanEngine.RemoveTasksByDeviceKey`	`feedbackPending` 残留未清理（泄漏）
R6	`ResourceController.ConnectionLimit`	`connectionCount` 无生产递增 → 该上限失效
R7	`omron/mitsubishi/snmp`	共享链路但未绑定 channelMu 且无 connMgr
R8	`SoakMonitor/GCMonitor` 循环	无 recover

4. 稳定性指标定义

指标全部可从 ScanEngineMetrics.Snapshot() / CircuitBreaker.Snapshot() / 通道诊断 API 读取。

4.1 调度连续性

指标	键名	定义
任务执行总数	`tasks_executed`	累计派发执行次数
成功率	`tasks_succeeded / tasks_executed`	采集成功占比
调度滞后 P95	`scan_lag_p95_ms`	实际开始时间 − 计划时间的 P95
扫描漂移均值	`scan_drift_avg_ms`	周期偏移均值
错过截止	`scan_miss_deadline_total`	超 `DeadlineAt` 才派发的次数
防饿死救援	`starvation_rescue_total`	触发救援次数（应长期为 0）

4.2 恢复能力（本次新增）

指标	键名	定义
采集 panic 次数	`task_panics_total`	必须告警，正常应恒为 0
任务重新武装次数	`task_recoveries_total`	panic/异常后恢复入队次数
熔断打开次数	`circuit_breaker_open_total`	设备级熔断触发数
熔断拒绝次数	`circuit_breaker_reject_total`	短路保护命中数
通道重连次数	`RecordReconnect`	看门狗 / 驱动触发重连数

4.3 数据一致性

指标	定义
点位质量分布	`Good / Uncertain / Bad` 占比
影子陈旧值兜底	`preservedShadowValue` 命中次数（Bad 时保留末次 Good）
点位降级数	`PointDegradationManager` 降级点位数
设备有效状态	`resolveEffectiveDeviceState` 的 Online/Unstable/Offline/Quarantine 分布

4.4 稳定性的三类量化口径

连续无故障运行时长：进程启动后 task_panics_total==0 且 starvation_rescue_total 增量==0 的最长连续时段（由 SoakMonitor 采样）。
异常恢复率 = 恢复成功次数 / 异常触发次数：
- 连接维度：RecordSuccess / (RecordSuccess + RecordFailure)（ConnectionManager）
- 任务维度：task_recoveries_total / task_panics_total（目标 100%，每个 panic 都重新武装）
- 熔断维度：HalfOpen 探测成功率
数据一致性：同一设备同一时刻，Shadow 值与最近一次 Good 采集一致（quality != Good 时值必须来自 preservedShadowValue，不得出现撕裂/中间态）。

5. 验收标准（可量化）

5.1 正确性（必须全绿）

#	条件	验证方式
A1	并发执行下无数据竞态	`go test -race ./internal/core/... ./internal/driver/...` 零报告
A2	无确定性死锁	全部新增死锁回归测试通过（ENIP / BACnet 重启 / 并发 Connect）
A3	防饿死不产生重复队条目	`TestAntiStarvation_RepeatedRescueKeepsSingleQueueEntry` 队列长度恒为 1
A4	任何驱动 panic 不导致进程退出	`TestExecutionLayer_RecoversDriverPanic` 通过
A5	panic 后任务不丢	`TestExecuteTaskAsync_RecoversPanicAndRearmsTask` 断言任务重新入队
A6	Params 并发读写无致命错误	`TestUpdateTaskDriverConfig_ConcurrentWithReaders`（`-race`）通过
A7	全量测试无回归	`go test -short ./internal/...` 全 PASS

5.2 稳定性（SLO）

指标	验收阈值
`task_panics_total`	= 0（任何增长视为 P0 缺陷）
`starvation_rescue_total` 增量	= 0（稳态 72h 内不增长）
`scan_lag_p95_ms`	≤ 100ms
`scan_drift_avg_ms`	≤ 50ms
`scan_miss_deadline_total` 增量	= 0
`circuit_breaker_reject_total` 增量	= 0（稳态）
通道重连成功率	≥ 99%（链路恢复后 30s 内自动回到 Good）
异常恢复率	= 100%（panic 必重新武装；断链必自动重连）

5.3 回归基线（本次实测）

项	结果
`internal/core`（跳过既有脆弱 HTTP 用例）	PASS
`internal/driver/...`（13 协议）	PASS
新增并发/恢复测试（`-race`）	PASS
EtherNet/IP 死锁复现测试（回滚修复时）	FAIL → 证明测试有效
防饿死重复入队复现（回滚修复时）	FAIL → 证明测试有效

6. 恢复机制设计（异常检测 → 自动重启 → 状态回滚 → 断线重连）

6.1 异常检测

检测器	位置	触发条件	动作
熔断器	`circuit_breaker.go`	连续 5 次超时或 60s 窗口失败率 > 40%（≥10 样本）	打开 30s → HalfOpen 探针
点位降级	`point_degradation_manager.go`	连续失败 ≥3	指数退避探测（5s→5min）
驱动点位冷却	`modbus/scheduler.go`	3 次→60s；10 次→5min；非法地址永久跳过（链路错误豁免）	跳过该点，不解冻设备
设备隔离	`bacnet/isolation.go`	连续 3 次失败	指数退避 1min→1h + 抖动
健康探针	`ReadPoints` 前置 `Health()`	`Health()!=Good`	触发重连 / 短路
panic 计数	`ExecutionLayer.readPoints`	驱动 panic	转为读错误 + `task_panics_total++`
通道健康巡检	`ChannelManager.channelHealthLoop`	链路 down 且超冷却	触发 `ScheduleReconnect`

6.2 自动重启（自愈）

层级	机制
任务级	panic 后 `rearmTaskAfterPanic`：置 Idle → 重排 → 重新入队（`task_recoveries_total++`）
调度级	`safeProcessReadyTasks`：recover + 队列重建，调度协程永不死亡
连接级	`ConnectionManager.EnsureConnected`：指数退避 + 冷却 + 全局限流（10 次/秒）
通道级	健康看门狗 15s 巡检 → `ScheduleReconnect`
进程级	systemd `Restart=always / RestartSec=5`（最后兜底）

6.3 状态回滚

场景	回滚动作
调度循环 panic，堆结构可能损坏	`rebuildQueueAfterPanic`：从权威 `tasks` map 重建优先队列，丢弃重复/悬垂条目，重算 `queued`
任务 panic	状态回退为 Idle 并重置调度锚点
采集失败	`preservedShadowValue` 保留末次 Good 值 + `quality=Bad`，避免下游读撕裂值
配置变更	Params copy-on-write：旧引用保持自洽，读者不会读到半更新配置
点位配置变更	Modbus 自动清除永久跳过/冷却

6.4 断线重连

链路异常
   │
   ├─ 驱动自检测（失败计数达阈值）──► ScheduleReconnect
   ├─ 通道看门狗（Health()==Bad）───► ScheduleReconnect   ← 本次新增
   ├─ 采集收尾链路错 ────────────────► markChannelDevicesOffline + 设备级恢复探针
   └─ StartChannel 连接失败 ─────────► markChannelDevicesOffline + ScheduleReconnect
                                          │
                                          ▼
                              ConnectionManager.EnsureConnected
                       （退避 → 冷却 → 全局限流 → linkMu 串行化拨号）
                                          │
                                          ▼
                                     RecordSuccess → 恢复

串行化保证：runConnect 持 channelMu（linkMu）执行 connectOnce，与扫描 I/O、API 写互斥，避免”重连拨号与在用 socket 打架”。要求驱动实现 drv.LinkMutexBinder 完成注入。

6.5 参考 Kepware 的简化

Kepware 的核心做法是 “把复杂度从热路径赶到配置/诊断层”，本次据此做了三处简化：

冷却/降级阈值集中在驱动内：Modbus PointScheduler 用固定阈值表（3 次→60s，10 次→5min）+ 一个哨兵时间 permanentSkipUntil 表达”永久跳过”，不引入状态机对象。
合并状态视图：不新增统一恢复状态机（避免 5 套枚举互转的复杂度），而是保留各层既有状态，通过指标 + 看门狗做单向驱动——恢复动作只依赖 Health() 与失败计数两个简单信号。
队列成员位替代“查重”：用 ScanTask.queued 一个布尔位取代”每次救援时线性扫描队列查重”，把 O(n) 变成 O(1)，与 Kepware 用”标签状态位”避免重复入队同构。

7. 变更清单（本次实现）

文件	变更
`internal/core/scan_engine.go`	`ScanTask.queued` + `isQueued/setQueued`；`paramsSnapshot/setParams`；`taskShadowChannelIDLocked`/`taskDegradeOnFailureLocked`；防饿死改 `heap.Fix`；`RemoveTask` 清队列；`executeTaskAsync` recover + `rearmTaskAfterPanic`；`safeProcessReadyTasks` + `rebuildQueueAfterPanic`；`safeSLAWarnings`
`internal/core/execution_layer.go`	`readPoints` 加 panic recover（命名返回值）；Params 读取走 `paramsSnapshot`
`internal/core/scan_engine_metrics.go`	新增 `task_panics_total` / `task_recoveries_total`
`internal/core/channel_manager.go`	新增通道健康看门狗（`channelHealthLoop` / `checkChannelHealth` / `safeCheckChannelHealth`）
`internal/driver/ethernetip/transport.go`	新增 `resetConnection`，修复重连自死锁
`internal/driver/opcua/opcua.go`	重连/懒连网络 RPC 移出 `d.mu`
`internal/core/concurrency_invariants_test.go`	新增 7 个并发不变量/恢复测试
`internal/core/channel_health_watchdog_test.go`	新增 2 个看门狗测试
`internal/driver/ethernetip/reconnect_deadlock_test.go`	新增 2 个重连死锁回归测试

8. 后续建议（按优先级）

补齐共享链路绑定：为 omron-fins / mitsubishi-slmp / snmp 实现 LinkMutexBinder（必要时引入 ConnectionManager），消除 R7。
修复 R1–R5：BACnet Health() 加锁、connectOnce 锁序对齐 Disconnect、NodeStatus 保护一致化、DataPipeline 关闭竞态、feedbackPending 清理。
恢复状态持久化：熔断/降级/隔离内存态重启即丢，建议落 bbolt 以支持”重启后维持隔离梯度”。
进程级看门狗：为关键循环统一加 recover 后，可增加”心跳文件 + systemd watchdog”实现进程级卡死检测。